package com.sunlands.deskmate.message.handler.receiver.impl;

import com.sunlands.deskmate.channel.ChannelInfo;
import com.sunlands.deskmate.channel.ChannelManager;
import com.sunlands.deskmate.config.RabbitMqConfig;
import com.sunlands.deskmate.message.entity.Messages;
import com.sunlands.deskmate.message.handler.receiver.Receiver;
import com.sunlands.deskmate.message.handler.sender.trigger.SendMessageTrigger;
import com.sunlands.deskmate.message.handler.sender.trigger.TriggerEvent;
import com.sunlands.deskmate.message.handler.sender.trigger.TriggerEventEnum;
import com.sunlands.deskmate.message.handler.sender.trigger.UnsyncMessageAcquire;
import com.sunlands.deskmate.message.handler.storager.unsync.UnsyncMessageStorager;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.util.Pair;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static com.sunlands.deskmate.config.RabbitMqConfig.EXCHANGE_SUCCESS;
import static com.sunlands.deskmate.message.enums.MessageType.NOTICE_EVENT;

/**
 * @author anjunling@sunlands.com
 */
@Slf4j
@Component
public class AckMessageReceiver implements Receiver {

    private static final ExecutorService executor = Executors.newFixedThreadPool(Math.max(4, Runtime.getRuntime().availableProcessors() * 2));
    @Autowired
    private UnsyncMessageStorager storager;
    @Autowired
    private SendMessageTrigger sendDispatcher;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private UnsyncMessageAcquire unsyncMessageAcquire;
    @Autowired
    private RabbitMqConfig rabbitMqConfig;

    @Override
    public void receive(Channel channel, Messages message) {

        log.info("11、回执消息，删除未读库记录 AckMessageReceiver message={}", message);
        Long messageId = message.getAckId();
        if (Objects.isNull(messageId)) {
            return;
        }

        //删除已同步消息
        ChannelInfo channelInfo = ChannelManager.getChannelInfo(channel);
        Pair<ChannelInfo, Long> channelInfoMessageIdPair = Pair.of(channelInfo, messageId);

        CompletableFuture.completedFuture(channelInfoMessageIdPair).thenAcceptAsync(pair -> {
            ChannelInfo chInfo = pair.getFirst();
            String sessionKey = chInfo.getSessionKey();

            //未发送消息列表
            List<Messages> list = unsyncMessageAcquire.getUnreadMessage(sessionKey, 0L, messageId);

            if (list.isEmpty()) {
                log.debug("消息以及ack，不需要重复ack {}", messageId);
                return;
            }
            long msgId = pair.getSecond();
            storager.setLatestReadMessageId(sessionKey, msgId);

            //解锁
            TriggerEvent dispatchEvent = new TriggerEvent(TriggerEventEnum.UNSYNC, chInfo);
            //释放锁，继续发送下面的消息
            sendDispatcher.epochNext(dispatchEvent);
            if (!CollectionUtils.isEmpty(list)) {
                list.forEach(msg -> {
                    if (!NOTICE_EVENT.getType().equals(msg.getType())) {
                        rabbitTemplate.convertAndSend(EXCHANGE_SUCCESS, rabbitMqConfig.getEnv(), msg);
                    }
                });
            }
            log.debug("ack messages getSessionKey={} msgId={}", sessionKey, msgId);
        }, executor);
    }

}
